package com.asiainfo.indb.control.impl;

import com.asiainfo.indb.control.FileCutter;
import com.asiainfo.indb.control.InDbService;
import com.asiainfo.indb.control.JavaInDbTask;
import com.asiainfo.indb.control.LogService;
import com.asiainfo.indb.control.db.DbTool;
import com.asiainfo.indb.model.InDbGather;
import com.asiainfo.indb.view.TablePanel;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Java多线程入库模式
 *
 * @author zym
 * @date 2019-10-18 17:20
 */
public class JavaDivisiveInDbService implements InDbService {

    private ExecutorService threadPool;

    @Override
    public void stop() {
        if (threadPool != null) {
            threadPool.shutdownNow();
        }
    }

    @Override
    public void indb(InDbGather gather, LogService logService) throws IOException {
        logService.report("入库前校验...");

        // 文件检查
        Path filePath = fileCheck(gather, logService);

        // 数据库和表检查
        String[] tbAndCols = null;
        if (filePath != null) {
            tbAndCols = dbCheck(gather, logService);
        }

        // 文件拆分
        // 文件入库
        int tbAndColsLen = 2;
        if (tbAndCols != null && tbAndCols.length == tbAndColsLen) {
            long size = filePath.toFile().length();
            if (size > 0) {
                // 获取基本控制参数
                int skip = 0;
                String skipStr = gather.get("SKIP_NUMBER");
                if (StringUtils.isNumeric(skipStr)) {
                    skip = Integer.parseInt(skipStr);
                }

                // 编码格式
                String fileEncode = gather.get("FILE_ENCODE");

                // 并发数
                int parallel = 1;
                String parallelStr = gather.get("PARALLEL");
                if (StringUtils.isNumeric(parallelStr)) {
                    parallel = Integer.parseInt(parallelStr);
                    if (parallel <= 0) {
                        parallel = 1;
                    }
                }
                logService.report("当前并行数：" + parallel);

                // 初始化线程池
                this.threadPool = createThreadPool(parallel);
                final FileCutter cutter = new FileCutter(skip, true, fileEncode, filePath, parallel, logService);

                // 提交执行
                logService.beginInDb(parallel);
                for (int i = 0; i < parallel; i++) {
                    this.threadPool.submit(new JavaInDbTask(cutter, logService, gather));
                }
                logService.endInDb();
            } else {
                logService.report("文件大小异常，请确认文件状态");
            }
        }
    }

    /**
     * 初始化线程池
     *
     * @param parallel 并行数
     * @return 线程池
     */
    private ExecutorService createThreadPool(int parallel) {
        return new ThreadPoolExecutor(parallel, parallel, 5,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(parallel), new ThreadFactory() {
            private AtomicInteger counter = new AtomicInteger();

            @Override
            public Thread newThread(Runnable task) {
                Thread thread = new Thread(task, "InDbService-" + counter.incrementAndGet());
                thread.setDaemon(true);
                return thread;
            }
        });
    }

    /**
     * 数据库和表检查，以及列检查
     *
     * @param data 表数据
     * @param log  日志
     * @return 表名和需要入库对应的列名
     */
    private String[] dbCheck(InDbGather data, LogService log) {
        String tb = data.get("TABLE_NAME");
        String cols = data.get("TABLE_COLS");

        try {
            log.report("开始数据库和表列有效性校验...");
            List<String> tabCols = DbTool.getTableCols(tb, data);
            if (tabCols != null && StringUtils.isNotBlank(cols)) {
                for (String col : cols.split(TablePanel.COL_SPLIT)) {
                    if (!tabCols.contains(col)) {
                        throw new IllegalArgumentException("表" + tb + "没有名为" + col + "的列");
                    }
                }

                return new String[]{tb, cols};
            } else {
                log.report("入库列为空，请添加");
            }
        } catch (SQLException e) {
            // 校验失败
            log.report("数据库和表信息校验失败：" + e.getMessage());
        }


        return null;
    }

    /**
     * 文件检查
     *
     * @param data 文件数据
     * @param log  日志
     * @return 文件地址
     */
    private Path fileCheck(InDbGather data, LogService log) {
        String file = data.get("FILE");
        log.report("开始文件校验：" + file);
        if (StringUtils.isBlank(file)) {
            log.report("入库文件为空，终止！");
            return null;
        }

        Path path = new File(file).toPath();
        if (!Files.exists(path)) {
            log.report("入库文件不存在，终止！");
            return null;
        }

        return path;
    }
}
